package com.csw.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

//4、统计偏科最严重的前100名学生  [学号，姓名，班级，科目，分数
object Text6 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("text1")

    val sc: SparkContext = new SparkContext(conf)

    //读取数据
    val scoresRDD: RDD[String] = sc.textFile("spark/data/score.txt")

    val studentsRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    //获取id和score的kv集合
    val scoresIdScoreRDD: RDD[(String, Int)] = scoresRDD.map(i => (i.split(",")(0), i.split(",")(2).toInt))

    //求每一个id的avg(score)
    val scoresAvgRDD: RDD[(String, Int)] = scoresIdScoreRDD.reduceByKey((x, y) => {
      (x + y) / 6
    })
    //    //把avg(score)转换为一个数组
    //    val a: RDD[Int] = scoresAvgRDD.map(kv=>kv._2)
    //    val AvgScore: Array[Int] = a.collect()

    //对scoresIdScoreRDD进行分组
    val scoresGroupRDD: RDD[(String, Iterable[Int])] = scoresIdScoreRDD.groupByKey()

    //对分组后的rdd进行map得到id和6门学科成绩
    val scoresGroupMapRDD: RDD[(String, (Int, Int, Int, Int, Int, Int))] = scoresGroupRDD.map(kv => {
      val id: String = kv._1
      val list: List[Int] = kv._2.toList
      val score1: Int = list(0)
      val score2: Int = list(1)
      val score3: Int = list(2)
      val score4: Int = list(3)
      val score5: Int = list(4)
      val score6: Int = list(5)
      (id, (score1, score2, score3, score4, score5, score6))
    })

    //把scoresAvgRDD和scoresGroupMapRDD关联
    val joinRDD: RDD[(String, ((Int, Int, Int, Int, Int, Int), Int))] = scoresGroupMapRDD.join(scoresAvgRDD)


    //通过公式计算方差
    val sRDD: RDD[(String, Double)] = joinRDD.map(i => {
      val id: String = i._1
      val ss: Double = ((i._2._1._1 - i._2._2) ^ 2 + (i._2._1._2 - i._2._2) ^ 2 + (i._2._1._3 - i._2._2) ^ 2 + (i._2._1._4 - i._2._2) ^ 2 + (i._2._1._5 - i._2._2) ^ 2) / 6.toDouble
      (id, ss)
    })

    //排序求偏科最严重的前100名学生
    val sSort: Array[(String, Double)] = sRDD.sortBy(i => i._2, false).take(100)

    //求出偏科最严重的前100名学生id
    val getid: Array[String] = sSort.map(i => i._1)

    //对成绩表进行转换
    val scoresMapRDD: RDD[(String, String)] = scoresRDD.map(i => {
      val id: String = i.split(",")(0)
      val subjectId: String = i.split(",")(1)
      val score: String = i.split(",")(2)

      (id, subjectId + "," + score)
    })
    //过滤得到偏科最严重的前100名学生id,科目，成绩
    val scoresMapFilterRDD: RDD[(String, String)] = scoresMapRDD.filter(kv => {
      val id: String = kv._1
      getid.contains(id)
    })
    //先对student表做处理，取出要显示的数据和学生表关联偏科最严重的前100名学生
    val studentsMapRDD: RDD[(String, String)] = studentsRDD.map(i => {
      val id: String = i.split(",")(0)
      val name: String = i.split(",")(1)
      val clazz: String = i.split(",")(4)
      (id, name + "," + clazz)
    })

    val result: RDD[(String, (String, Option[String]))] = scoresMapFilterRDD.leftOuterJoin(studentsMapRDD)

    result.foreach(println)
    println(result.count())
  }
}
